package com.example.kafka.controller;

import com.alibaba.fastjson2.JSONObject;
import com.example.kafka.service.KafkaLogService;
import com.example.kafka.service.KafkaProducerService;
import com.example.kafka.service.KafkaTopicService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.TopicDescription;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.Map;
import java.util.Set;

@RestController
@RequestMapping("/kafka")
@Slf4j
public class KafkaController {

    @Autowired
    KafkaProducerService producerService;

    @Autowired
    private KafkaTopicService topicService;

    @Autowired
    private KafkaLogService logService;

    @GetMapping("/send/{message}")
    public String sendMessage(@PathVariable String message) {
        producerService.sendMessage("default-topic", message);
        return "Message sent: " + message;
    }

    @GetMapping("/send")
    public String sendMessage() {
        JSONObject json = new JSONObject();
        json.put("level", "ERROR");
        producerService.sendMessage("application-logs", json.toString());
        return "Message sent: " + json;
    }

    /**
     * 发送消息
     */
    @GetMapping("/send/{topic}/{key}/{message}")
    public String sendMessage(@PathVariable String topic, @PathVariable String key, @PathVariable String message) {
        producerService.sendMessageWithCallback(topic, key, message);
        return "topic sent: " + topic + "，key sent: " + key + "，Message sent: " + message;
    }

    /**
     * 创建主题
     *
     * @param name       主题名称
     * @param partitions 分区
     * @param replicas   副本
     */
    @PostMapping("/create")
    public String create(@RequestParam String name,
                         @RequestParam(defaultValue = "1") int partitions,
                         @RequestParam(defaultValue = "1") short replicas) {
        try {
            topicService.createTopic(name, partitions, replicas);
            return "✅ 主题创建成功：" + name;
        } catch (Exception e) {
            return "❌ 创建失败：" + e.getMessage();
        }
    }

    /**
     * 删除主题
     * @param name 主题名称
     */
    @DeleteMapping("/delete")
    public String delete(@RequestParam String name) {
        try {
            topicService.deleteTopic(name);
            return "✅ 主题删除成功：" + name;
        } catch (Exception e) {
            return "❌ 删除失败：" + e.getMessage();
        }
    }

    /**
     * 列出所有主题
     */
    @GetMapping("/list")
    public Set<String> list() throws Exception {
        return topicService.listTopics();
    }

    /**
     * 描述主题
     */
    @GetMapping("/describe")
    public Map<String, TopicDescription> describe(@RequestParam String name) throws Exception {
        return topicService.describeTopic(name);
    }

    /**
     * 查询用户信息
     */
    @GetMapping("/{userId}")
    public String getUserInfo(@PathVariable String userId, @RequestParam String query) {
        log.info("查询用户信息，用户ID: {}, 条件: {}", userId, query);
        return logService.queryUserInfo(userId, query);
    }

}
